Remove DRPC per-function request queues once they are empty - #8985

Disclaimer: this comment was generated with the help of an LLM.

The new test proves the map returns to empty (the leak fix), but it's essentially sequential — it wouldn't catch a lost-request race. Since correctness here depends on every queue mutation going through the per-key compute lock, a concurrent stress test would strengthen it. Rough sketch:

@Test
public void concurrentSubmitAndFetchLosesNoRequests() throws Exception {
    try (DRPC server = new DRPC(new StormMetricsRegistry(), null, 60_000)) {
        int n = 10_000;
        for (int i = 0; i < n; i++) {
            exec.submit(() -> server.executeBlocking("fn", "x"));
        }
        int served = 0;
        while (served < n) {
            DRPCRequest req = server.fetchRequest("fn");
            if (req != null && !req.get_request_id().isEmpty()) {
                server.returnResult(req.get_request_id(), "ok");
                served++;
            }
        }
        assertEquals(n, served);                          // none lost / duplicated
        assertEquals(0, server.getNumTrackedFunctions()); // no queue left behind
    }
}

Not a blocker — just closes the gap on the part of the change that's hardest to get right.

Disclaimer: this comment was generated with the help of an LLM.

The new test proves the map returns to empty (the leak fix), but it's essentially sequential — it wouldn't catch a lost-request race. Since correctness here depends on every queue mutation going through the per-key compute lock, a concurrent stress test would strengthen it. Rough sketch:

@Test
public void concurrentSubmitAndFetchLosesNoRequests() throws Exception {
    try (DRPC server = new DRPC(new StormMetricsRegistry(), null, 60_000)) {
        int n = 10_000;
        for (int i = 0; i < n; i++) {
            exec.submit(() -> server.executeBlocking("fn", "x"));
        }
        int served = 0;
        while (served < n) {
            DRPCRequest req = server.fetchRequest("fn");
            if (req != null && !req.get_request_id().isEmpty()) {
                server.returnResult(req.get_request_id(), "ok");
                served++;
            }
        }
        assertEquals(n, served);                          // none lost / duplicated
        assertEquals(0, server.getNumTrackedFunctions()); // no queue left behind
    }
}

Not a blocker — just closes the gap on the part of the change that's hardest to get right.

Added something in that way. Thanks